package com.partsoft.dits.integrate;



import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import javax.sql.DataSource;

import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import org.springframework.util.ObjectUtils;

import com.partsoft.dits.AbstractMessageEntityBatchProcessor;
import com.partsoft.dits.DataEdgeDescriptor;
import com.partsoft.dits.DataIntegrateBatch;
import com.partsoft.dits.DataReceiver;
import com.partsoft.dits.DescriptorAccepter;
import com.partsoft.dits.EdgeDescriptor;
import com.partsoft.dits.TableDataEdgeDescriptor;
import com.partsoft.dits.dsm.DataSourceFactory;
import com.partsoft.dits.log.ChangeLogStorage;
import com.partsoft.dits.log.Log;
import com.partsoft.dits.message.DataProperty;
import com.partsoft.dits.message.MessageDataItem;
import com.partsoft.dits.message.MessageDataItemVisitor;
import com.partsoft.dits.message.MessageEntity;
import com.partsoft.dits.message.MessageEntityVisitor;
import com.partsoft.dits.message.MessageException;
import com.partsoft.dits.message.MessageHeader;
import com.partsoft.dits.message.MessageProcessor;
import com.partsoft.dits.transform.TransformException;
import com.partsoft.dits.transform.TransformReactor;
import com.partsoft.dits.utils.DbUtils;
import com.partsoft.dits.utils.EdgeLogUtils;
import com.partsoft.dits.utils.MessageEntityUtils;

@SuppressWarnings("unchecked")
public abstract class AbstractTableDataMessageReceiver implements DataReceiver {

	private ChangeLogStorage logStorage;

	private AbstractMessageEntityBatchProcessor byteMsgBatchHelper = new AbstractMessageEntityBatchProcessor() {
		@Override
		protected Object doProcess(DataEdgeDescriptor dataDescriptor, MessageEntity bytes,
				DescriptorAccepter<EdgeDescriptor> accepter) {
			try {
				AbstractTableDataMessageReceiver.this.doProcess((TableDataEdgeDescriptor) dataDescriptor, bytes);
			} catch (MessageException e) {
				throw new IllegalArgumentException(e.getMessage(), e);
			}
			return null;
		}

		protected void validateBatch(DataIntegrateBatch batch) {
			super.validateBatch(batch);
			validateSupportDataDescriptor(batch.getDataDescriptor());
		};

	};

	public abstract TransformReactor getTransformReactor();

	public abstract DataSourceFactory getDataSourceFactory();

	public abstract MessageProcessor getMessageProcessor();

	public abstract TableDataDescriptorRegistry<TableDataEdgeDescriptor> getDataDescriptorRegistry();

	public void setLogStorage(ChangeLogStorage logStorage) {
		this.logStorage = logStorage;
	}

	public ChangeLogStorage getLogStorage() {
		return logStorage;
	}

	public boolean supportData(Class<? extends DataEdgeDescriptor> descriptorClazz) {
		return ClassUtils.isAssignable(TableDataEdgeDescriptor.class, descriptorClazz);
	}

	public boolean supportData(DataEdgeDescriptor dataDescriptor) {
		if (!ClassUtils.isAssignableValue(TableDataEdgeDescriptor.class, dataDescriptor))
			return false;
		Assert.notNull(getDataDescriptorRegistry(), "请设置数据描述符注册表。");
		return supportData((Class<? extends DataEdgeDescriptor>) ClassUtils.getUserClass(dataDescriptor)) ? getDataDescriptorRegistry()
				.hasDescriptor((TableDataEdgeDescriptor) dataDescriptor) : false;
	}

	protected void validateSupportDataDescriptor(DataEdgeDescriptor descriptor) {

	}

	public void process(DataIntegrateBatch batch) {
		byteMsgBatchHelper.process(batch);
	}

	protected void doProcess(TableDataEdgeDescriptor dataDescriptor, MessageEntity msgEntity) throws MessageException {
		Assert.notNull(getDataDescriptorRegistry(), "必须设置数据描述符注册表。");
		Assert.notNull(getMessageProcessor(), "必须设置消息处理器。");
		Assert.notNull(getDataSourceFactory(), "必须设置连接工厂。");
		Assert.notNull(msgEntity, "必须传入消息实体。");
		Assert.isTrue(getDataSourceFactory().hasDataSource(dataDescriptor.getDataSourceName()),
				String.format("找不到表数据对应的数据源(%s)", dataDescriptor.getDataSourceName()));
		Assert.isTrue(ObjectUtils.nullSafeEquals(msgEntity.getHeader().getMessageCode(),
				dataDescriptor.getMessageCode()), String.format("数据描述符的消息(%s)与解码后的消息(%s)不一致不能处理。",
				dataDescriptor.getMessageCode(), msgEntity.getHeader().getMessageCode()));

		if (Log.isDebugEnabled()) {
			Log.debug(String.format("开始处理消息(%s.%s)", msgEntity.getHeader().getMessageCode(), msgEntity.getHeader()
					.getMessageId()));
		}
		doProcessReceiveMessage(dataDescriptor, msgEntity);
	}

	private void fillProcessReceiveMessageContext(ReceiveMessageContext ctx, TableDataEdgeDescriptor dataDescriptor) {
		if (dataDescriptor.isConversion())
			ctx.addTransformScript(dataDescriptor, dataDescriptor.getTransformScript());
		if (!ctx.hasDataSource(dataDescriptor.getDataSourceName())) {
			ctx.addDataSource(dataDescriptor.getDataSourceName(),
					getDataSourceFactory().getDataSource(dataDescriptor.getDataSourceName()));
		}
		DataSource ds = ctx.getDataSource(dataDescriptor.getDataSourceName());
		if (!ctx.hasTableColumns(dataDescriptor)) {
			ctx.addTableColumns(dataDescriptor,
					DbUtils.getTableColumns(ds, dataDescriptor.getSchemaName(), dataDescriptor.getTableName()));
		}
		ctx.addReceiveBean(dataDescriptor.getMessageCode(), dataDescriptor);
	}

	private void doProcessReceiveMessage(TableDataEdgeDescriptor descriptor, final MessageEntity msgEntity)
			throws MessageException {
		final ReceiveMessageContext ctx = new ReceiveMessageContext(getMessageProcessor(), getDataSourceFactory(),
				getTransformReactor());
		fillProcessReceiveMessageContext(ctx, descriptor);

		MessageEntityUtils.traverseMessage(msgEntity, new MessageEntityVisitor() {
			public void visitMessageEntity(MessageEntity entity) throws MessageException {
				if (ObjectUtils.nullSafeEquals(entity, msgEntity))
					return;
				MessageHeader msgHeader = entity.getHeader();
				if (!getDataDescriptorRegistry().hasMessageCode(msgHeader.getMessageCode())) {
					throw new MessageException(String.format("消息代码(%s)尚未定义接收", msgHeader.getMessageCode()), entity);
				}
				ctx.addReceiveBeans(msgHeader.getMessageCode(),
						Arrays.asList(getDataDescriptorRegistry().getDescriptors(msgHeader.getMessageCode())));
				for (TableDataEdgeDescriptor bean : getDataDescriptorRegistry().getDescriptors(
						msgHeader.getMessageCode())) {
					fillProcessReceiveMessageContext(ctx, bean);
				}
			}
		});

		// 开始联合事务
		ctx.startContextTransaction();
		try {

			for (MessageDataItem dataitem : msgEntity.getDatas()) {
				for (MessageEntity parentMsg : dataitem.getParents()) {
					MessageEntityUtils.traverseMessage(parentMsg, new MessageDataItemVisitor() {
						public void visitDataItem(MessageDataItem dataItem) throws MessageException {
							try {
								saveMessageDataItem(ctx, dataItem);
							} catch (Exception e) {
								throw new MessageException(String.format("处理外键消息数据(%s)收发生错误:%s", dataItem,
										e.getMessage()), e);
							}
						}
					});
				}
				saveMessageDataItem(ctx, dataitem);
				for (MessageEntity childMsg : dataitem.getChilds()) {
					MessageEntityUtils.traverseMessage(childMsg, new MessageDataItemVisitor() {
						public void visitDataItem(MessageDataItem dataItem) throws MessageException {
							try {
								saveMessageDataItem(ctx, dataItem);
							} catch (Exception e) {
								throw new MessageException(String.format("处理子级消息数据(%s)收发生错误:%s", dataItem,
										e.getMessage()), e);
							}
						}
					});
				}
			}
			// 提交联合事务
			ctx.commitContextTransaction();
		} catch (Throwable e) {
			// 回滚联合事务
			ctx.rollbackContextTransaction();
			throw new MessageException(e.getMessage(), msgEntity, e);
		}

	}

	// 运行在独立的上下文中
	protected void saveMessageDataItem(ReceiveMessageContext ctx, MessageDataItem dataItem) throws MessageException,
			TransformException {
		Assert.notNull(ctx);
		Assert.notNull(dataItem.getOwner());
		MessageHeader msgHeader = dataItem.getOwner().getHeader();
		Assert.notNull(msgHeader);

		MessageProcessor messageProcessor = ctx.getMessageProcessor();
		TransformReactor transformReactor = ctx.getTransformReactor();

		if (!ctx.hasReceiveDefined(msgHeader.getMessageCode()))
			throw new MessageException(String.format("消息代码(%s)尚未定义接收", msgHeader.getMessageCode()), dataItem.getOwner());
		List<TableDataEdgeDescriptor> baseTableIntegrateEdgeDescriptors = new ArrayList<TableDataEdgeDescriptor>(
				ctx.getReceiveBeans(msgHeader.getMessageCode()));

		for (TableDataEdgeDescriptor baseTableIntegrateEdgeDescriptor : baseTableIntegrateEdgeDescriptors) {
			DataProperty targetColumns[] = ctx.getTableColumns(baseTableIntegrateEdgeDescriptor);
			List<Map<DataProperty, String>> newItems = new ArrayList<Map<DataProperty, String>>(1);
			newItems.add(dataItem.getValues());
			MessageEntity newMsg = messageProcessor.newMessage(null, msgHeader.getMessageCode(),
					msgHeader.getDefaultAction(), newItems, null);
			try {
				if (baseTableIntegrateEdgeDescriptor.isConversion()) {
					// newMsg =
					// transformReactor.validateTransformMessageEntity(newMsg,
					// baseTableIntegrateEdgeDescriptor.getTransformScript(),
					// targetColumns);
					newMsg = transformReactor.transform(newMsg, baseTableIntegrateEdgeDescriptor.getTransformScript());
					if (newMsg.hasData()) {
						MessageEntityUtils.validateMessageDataItemExpectColumns(newMsg, targetColumns, true);
					} else {
						EdgeLogUtils.info(baseTableIntegrateEdgeDescriptor, "转换后的消息无行数据，原始行数据：" + dataItem.toString());
						continue;
					}
				} else if (newMsg.hasData()) {
					MessageEntityUtils.validateMessageDataItemExpectColumns(newMsg, targetColumns);
				}
			} catch (MessageException e) {
				throw new MessageException(String.format("检查消息(%s)时发生错误:%s", baseTableIntegrateEdgeDescriptor,
						e.getMessage()), newMsg, e);
			}
			if (!newMsg.hasData()) {
				EdgeLogUtils.info(baseTableIntegrateEdgeDescriptor, "消息无行数据，原始行数据：" + dataItem.toString());
				continue;
			}
			for (MessageDataItem targetDataItem : newMsg.getDatas()) {
				DataProperty[] keyProps = baseTableIntegrateEdgeDescriptor.getKeyColumns();
				Map<DataProperty, String> itemValues = new HashMap<DataProperty, String>(targetDataItem.getValues());
				Map<DataProperty, String> keyValues = new HashMap<DataProperty, String>(keyProps.length);
				StringBuilder consKeyValue = new StringBuilder(24 * keyProps.length);
				boolean consKeyAppended = false;
				for (DataProperty dProps : keyProps) {
					String keyValue = itemValues.get(dProps);
					keyValues.put(dProps, itemValues.get(dProps));
					//itemValues.remove(dProps); by 赵岳峰，不能删除，否则新增时不能入库
					//TODO 此处可能需要调整分隔符号
					if (consKeyAppended) consKeyValue.append("|");
					consKeyAppended = true;
					consKeyValue.append(keyValue);
				}
				if (itemValues.size() == 0 || (itemValues.containsKey(DataProperty.DATA_ACTION) && itemValues.size() == 1)) {
					continue;
				}

				try {
					DbUtils.saveRecordToTable(ctx.getDataSource(baseTableIntegrateEdgeDescriptor.getDataSourceName()),
							baseTableIntegrateEdgeDescriptor.getSchemaName(),
							baseTableIntegrateEdgeDescriptor.getTableName(), dataItem.getAction(), keyValues, itemValues);
					if (getLogStorage() != null) {
						getLogStorage().saveChangeLog(baseTableIntegrateEdgeDescriptor, 
								consKeyValue.toString(), dataItem.getAction());
					}
				} catch (Throwable e) {
					Log.error(String.format("保存消息(%s)数据至数据库时发生错误:%s", 
							baseTableIntegrateEdgeDescriptor, e.getMessage()), e);
					throw new MessageException(String.format("保存消息(%s)时发生错误：%s", 
							baseTableIntegrateEdgeDescriptor, e.getMessage()), newMsg, e);
				}
			}
		}
	}

	public void receiveData(DataEdgeDescriptor dataDescriptor, MessageEntity databytes) {
		Assert.notNull(dataDescriptor);
		Assert.notNull(databytes);
		validateSupportDataDescriptor(dataDescriptor);
		try {
			doProcess((TableDataEdgeDescriptor) dataDescriptor, databytes);
		} catch (MessageException e) {
			throw new IllegalArgumentException(e.getMessage(), e);
		}
	}

	public void receiveData(MessageEntity msgEntity) {
		receiveData(msgEntity, null);
	}

	public void receiveData(MessageEntity msgEntity, DescriptorAccepter<EdgeDescriptor> accepter) {
		TableDataEdgeDescriptor tbs[] = getDataDescriptorRegistry().getDescriptors(
				msgEntity.getHeader().getMessageCode());
		Assert.notEmpty(tbs, String.format("找不到消息(%s)对应的接收描述符。", msgEntity.getHeader().getMessageCode()));

		try {
			for (TableDataEdgeDescriptor tb : tbs) {
				if (accepter == null || accepter.accept(tb)) {
					doProcess(tb, msgEntity);
				}
			}
		} catch (MessageException e) {
			throw new IllegalArgumentException(e.getMessage(), e);
		}

	}
}
